package day04;

import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.net.URI;
import java.util.HashMap;
import java.util.Map;
import java.util.Scanner;

//SourceFunction(接口) --> ParallelSourceFunction(接口) 并行的
public class RichHdfsSourceFunctionWithMap extends RichSourceFunction<Map<String, String>> {

    Boolean isRunning=true;
    private String url=null;
    //每次读取文件保存最后状态的md5值
    private String md5=null;
    //每隔 时间扫描
    private static final int interval=3000;
    Path path=null;
    FileSystem fs=null;

    public RichHdfsSourceFunctionWithMap(String url) {
        //初始化执行的代码在driver端
        this.url =(url);
    }

    @Override
    public void open(org.apache.flink.configuration.Configuration parameters) throws Exception {
        Configuration conf = new Configuration();
        fs=FileSystem.get(new URI("hdfs://ns1"),conf,"yeniu");
        path = new Path(this.url);
        System.out.println("-->open");
    }

    @Override
    public void close() throws Exception {
        System.out.println("-->close");
        super.close();
    }

    //hdfs连接的时候需要指定配置文件 core-site.xml   hdfs-site.xml
    @Override //executor端
    public void run(SourceContext<Map<String, String>> ctx) throws Exception {
        while(isRunning){
            //如果不存在跳过当次
            if(!fs.exists(path)){
                Thread.sleep(interval);
                continue;
            }

            //获取校验核文件
            String fileChecksum = fs.getFileChecksum(path).toString();
            //获取文件md5()
            String current_md5 = fileChecksum.substring(fileChecksum.indexOf(":") + 1);

            if(this.md5==null||!current_md5.equals(this.md5)){
                //存在就读
                FSDataInputStream in = fs.open(path);
                Scanner sc= new Scanner(in);
                Map<String, String> map = new HashMap<>();
                while (sc.hasNext()){
                    String line = sc.nextLine();
                    String[] s = line.split(" ");
                    map.put(s[0],s[1]);
                }
                ctx.collect(map);
                //更新md5
                this.md5=current_md5;
                //关闭流
                in.close();
            }
            Thread.sleep(interval);

        }
    }

    @Override
    public void cancel() {
        System.out.println("-->cancel");
        isRunning=false;
    }
}
